﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Diagnostics.Contracts;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Actors;
using System.Threading.Actors.Utilities;
using System.Threading.Tasks;

namespace ActorGatewayService
{
    // Class that listens for client connection attempts to the actor gateway.
    public class ActorGatewayTcpListener
    {
        private readonly TcpListener m_listener;
        private readonly string m_fabricConnectionString;

        public ActorGatewayTcpListener(Uri uri, AddressFamily addressFamily = AddressFamily.InterNetwork, int serverPort = 19000)
        {
            bool useLocal = uri.Host.Contains("localhost") || ((addressFamily == AddressFamily.InterNetwork) ? uri.Host.Contains("127.0.0.1") : uri.Host.Contains("[::1]"));
            IPAddress ipAddr = null;
            if (useLocal)
                ipAddr = (addressFamily == AddressFamily.InterNetworkV6) ? IPAddress.IPv6Loopback : IPAddress.Loopback;
            else
                ipAddr = NetworkConfiguration.GetIPAddress(addressFamily);
            IPEndPoint ep = new IPEndPoint(ipAddr, uri.Port);
            m_listener = new TcpListener(ep);
            m_listener.Start();
            Endpoint = ep;
            m_fabricConnectionString = uri.Host + ":" + serverPort; // Need to plumb this through to GatewayClientHandler
            ListenerLoop(); // fire-and-forget
            MessageLogger.WriteLine("ActorGatewayTcpListener listening on IP address {0}, port {1}", ipAddr, uri.Port);
        }

        internal IPEndPoint Endpoint { get; private set; }

        public void Close()
        {
            MessageLogger.WriteLine("ActorGatewayTcpListener being explicitly closed...");
            m_listener.Stop();
        }

        private async void ListenerLoop()
        {
            try
            {
                while (true)
                {
                    var client = await m_listener.AcceptTcpClientAsync().ConfigureAwait(continueOnCapturedContext:false);
                    try
                    {
                        var handler = new GatewayClientHandler(client, m_fabricConnectionString);
                    }
                    catch (Exception e)
                    { 
                        // Don't halt the listener loop just because GCH handler throws
                        MessageLogger.WriteLine("GatewayClientHandler ctor({0}) threw exception: {1}", 
                            client.Client.RemoteEndPoint.ToString(), e.Message);
                    }
                }
            }
            catch (Exception e)
            { 
                // Listener probably got closed
                MessageLogger.WriteLine("ActorGatewayTcpListener.ListenerLoop threw: {0}", e.Message);
            }
        }
    }

    // Handles all operations for a client that gains access to the gateway.
    public class GatewayClientHandler
    {
        private TcpClient m_sourceClient;
        private EndPoint m_sourceEndpoint;
        private NetworkStream m_sourceStream;

        private Uri m_targetUri;
        private TcpClient m_targetClient;
        private NetworkStream m_targetStream;
        private String m_partitionName;

        // I could get this from System.Threading.Actors.SerializationHelper, but then
        // I would need to reference the json.net assembly.
        private static readonly Encoding PreferredEncoding = Encoding.UTF8;

        public GatewayClientHandler(TcpClient client, string fabricConnectionString)
        {
            m_sourceClient = client;
            m_sourceEndpoint = m_sourceClient.Client.RemoteEndPoint;
            // Without this, the client may delay sending data, in hopes that more data will
            // soon be coming.  I've decided that this is a bad idea for our current scenarios,
            // but we may want to revisit this decision for heavy streaming applications where
            // throughput is more important than turnaround-time.
            m_sourceClient.NoDelay = true;
            InitializeAsync(fabricConnectionString); // Don't want to block gateway listener.  
                               // Launch "fire-and-forget" async method for initialization chores.
        }

        private async void InitializeAsync(string fabricConnectionString)
        {
            try
            {
                m_sourceStream = m_sourceClient.GetStream();

                // Read in the target Uri
                var buffer = new byte[1024];
                var targetString = await NetworkHelpers.ReadStringAsync(m_sourceStream, buffer).ConfigureAwait(continueOnCapturedContext:false);
                // targetString may optionally contain a partition name, appended on with a # separator.
                if (targetString.Contains('#'))
                {
                    int index = targetString.IndexOf('#');
                    m_partitionName = targetString.Substring(index + 1);
                    targetString = targetString.Substring(0, index);
                }
                m_targetUri = new Uri(targetString);

                // Attempt to establish connection with the indicated target, harvesting a discriminator
                // string in the process.
                var tuple = await EstablishConnectionWithTargetAsync(fabricConnectionString);
                m_targetClient = tuple.Item1;
                string discriminator = tuple.Item2;
                if (m_targetClient == null)
                {
                    MessageLogger.WriteLine("GCH InitializeAsync: failed to connect to fabric cluster \"{0}\".  Try again later?", fabricConnectionString);
                    Shutdown();
                    return;
                }
                MessageLogger.WriteLine("GCH InitializeAsync: target actor = {0}{1}, remote = {2},{3}", 
                    targetString, 
                    (m_partitionName == null) ? "" : ", partition name = " + m_partitionName,
                    m_targetClient.Client.LocalEndPoint, m_targetClient.Client.RemoteEndPoint);

                if (m_targetClient != null)
                {
                    // See comments for m_sourceClient.NoDelay above.
                    m_targetClient.NoDelay = true;
                    m_targetStream = m_targetClient.GetStream();

                    // Send the discriminator & partition name and get a response
                    string discriminatorAndPartition = discriminator;
                    if (m_partitionName != null)
                        discriminatorAndPartition += "#" + m_partitionName;
                    NetworkHelpers.WriteString(discriminatorAndPartition, m_targetStream);
                    var status = await NetworkHelpers.ReadStringAsync(m_targetStream, buffer).ConfigureAwait(continueOnCapturedContext: false);
                    if (status.Equals("OK"))
                    {
                        NetworkHelpers.WriteString("OK", m_sourceStream);
                        SourceLoop(); // fire and forget
                        TargetLoop(); // fire and forget
                    }
                    else
                    {
                        var errString = String.Format("FAILED to connect with target {0}, partition name={1}, status={2}", 
                                                      targetString, m_partitionName ?? "null", status);
                        NetworkHelpers.WriteString(errString, m_sourceStream);
                        MessageLogger.WriteLine("GCH InitializeAsync({0}) failed on discrimination, partition name={1}, status={2}",
                                                targetString, m_partitionName ?? "null", status);
                        // Using the gateway, if a service is actively killed, sometimes a new service instance comes back up but
                        // we never resolve to it.  Some of the cache invalidation logic may not get a chance to run because we
                        // do not keep around a FabricActorClient in the same way.  So, explicitly try helping out by throwing away
                        // cached state.  (Note this state could be only temporarily incorrect, but this is just invalidating a cache.)
                        FabricServerClient.DumpCachedFabricClient();
                        Shutdown();
                    }
                }
                else
                {
                    NetworkHelpers.WriteString(String.Format("FAILED to establish connection with target {0} partition name={1}", 
                                                             targetString, m_partitionName ?? "null"), m_sourceStream);
                    MessageLogger.WriteLine("GCH InitializeAsync({0}) FAILED to establish a connection.  Shutting down.",
                        targetString);
                    Shutdown();
                }
            }
            catch (Exception e)
            {
                // In the case of an exception, shut everything down so that nobody is left hanging
                MessageLogger.WriteLine("GCH InitializeAsync({0},part={1}) threw: {2}.  Shutting everything down.  Full exception info: {3}",
                    m_targetUri == null ? "null" : m_targetUri.ToString(),
                    m_partitionName ?? "null",
                    e.Message, e);
                Shutdown();
            }
        }

        // Called after receiving the m_targetUri from the client.  Establishes a TCP connection with the target.
        private async Task<Tuple<TcpClient, string>> EstablishConnectionWithTargetAsync(string fabricConnectionString)
        {
            // Attempt to establish a tcp connection with the target
            var fsc = new FabricServerClient(new string[1] { fabricConnectionString });
            long? hashedPartitionKey = new Nullable<long>();
            if (m_partitionName != null)
                hashedPartitionKey = FabricServerClient.GetHashedPartitionKey(m_partitionName);
            // When connecting from a normal client to a local host, I found it necessary to wait at least 16 seconds,
            // if not 30 seconds to detect that a primary is down, wait for a new one to spin up, then listen on the right socket.
            // Most of that logic could live in ActorClient, but we can do some limited retry logic here.  Don't want to 
            // go overboard though (ie, 30 client retries * a large number of gateway retries is a lot of pinging).
            const int numRetries = 4;
            const int WaitTimeMS = 500;
            for (int i = 0; i<numRetries; i++)
            {
                var tuple = await fsc.TryResolveAndCreateRawClientAsync(m_targetUri, hashedPartitionKey, null);
                if (tuple.Item4 == FabricServerClient.FabricResolveReturnCode.Success)
                    return Tuple.Create(tuple.Item1, tuple.Item3);
                else if (tuple.Item4 == FabricServerClient.FabricResolveReturnCode.Retriable)
                {
                    Task.Delay(WaitTimeMS).Wait();
                    continue;
                }
                else
                {
                    Contract.Assert(tuple.Item4 == FabricServerClient.FabricResolveReturnCode.Fatal,
                        "Return code from TryResolveAndCreateRawClientAsync was not fatal!  Got: " + tuple.Item4);
                    return Tuple.Create((TcpClient)null, (string)null);
                }
            }
            return Tuple.Create((TcpClient)null, (string)null);
        }

        // Reads requests from the source and forwards them to the target
        private async void SourceLoop()
        {
            var buffer = new byte[1024];
            try
            {
                while (true)
                {
                    //MessageLogger.WriteLine("SourceLoop waiting on client...");
                    // We are reading requests from the source.
                    // Read the request packet and then forward it to the target.
                    var tuple = await NetworkHelpers.ReadRequestPacketAsync(m_sourceStream, buffer).ConfigureAwait(continueOnCapturedContext: false);
                    NetworkHelpers.WriteRequestPacket(m_targetStream, tuple.Item1, tuple.Item2);
                }
            }
            catch (Exception e)
            {
                HandleLoopException(e, "SourceLoop");
            }
        }

        // Reads responses/event publications from the target and forwards them to the source
        private async void TargetLoop()
        {
            var buffer = new byte[1024];
            try
            {
                while (true)
                {
                    //MessageLogger.WriteLine("TargetLoop waiting on target...");
                    // We are reading responses/events back from the target.
                    // Read the feedback packet and then forward it on to the source.
                    var tuple = await NetworkHelpers.ReadFeedbackPacketAsync(m_targetStream, buffer).ConfigureAwait(continueOnCapturedContext: false);
                    NetworkHelpers.WriteFeedbackPacket(m_sourceStream, tuple.Item1, tuple.Item2, tuple.Item3);
                }
            }
            catch (Exception e)
            {
                HandleLoopException(e, "TargetLoop");
            }
        }

        // Common logic for handling exceptions in SourceLoop or TargetLoop.
        private void HandleLoopException(Exception e, string whichLoop)
        {
            var exceptionSourceDescription = String.Format("GCH({0},{1},part={2}).{3}",
                    m_sourceEndpoint,
                    m_targetUri.ToString(),
                    m_partitionName ?? "null",
                    whichLoop);

            if (e is IOException)
            {
                var ioe = e as IOException;
                // The line has gone dead.  Let's shut everything down in both directions.
                Shutdown();
                if (ioe is EndOfStreamException)
                    MessageLogger.WriteLine("{0} reached end of stream.", exceptionSourceDescription);
                else
                    MessageLogger.WriteLine("{0} IOE: {1}.  Actively shutting down",
                        exceptionSourceDescription, ioe.Message);
            }
            else if (e is ObjectDisposedException)
            {
                // We used something after it had been disposed.  Consider this a benign race.
                MessageLogger.WriteLine("{0} ODE.  Passively shutting down.",
                    exceptionSourceDescription);
            }
            else
            {
                // We used something after it had been disposed.  Consider this a benign race.
                MessageLogger.WriteLine("{0} exception: {1}.  Passively shutting down.",
                    exceptionSourceDescription,
                    e.Message);
            }

        }

        // Called when something has gone wrong (typically a connection has gone dead)
        private void Shutdown()
        {
            lock (m_sourceEndpoint)
            {
                // close both TCP connections
                if (m_targetClient != null)
                {
                    m_targetClient.Close();
                    m_targetClient = null;
                }

                if (m_sourceClient != null)
                {
                    m_sourceClient.Close();
                    m_sourceClient = null;
                }

                // close all the streams

                if (m_sourceStream != null)
                {
                    m_sourceStream.Dispose();
                    m_sourceStream = null;
                }

                if (m_targetStream != null)
                {
                    m_targetStream.Dispose();
                    m_targetStream = null;
                }
            }
        }
    }

}
